package fun.tan90.easy.binlog.plugin.rocketmq.service.impl;

import cn.hutool.json.JSONUtil;
import fun.tan90.easy.binlog.plugin.common.service.PluginService;
import fun.tan90.easy.binlog.plugin.rocketmq.constants.RocketmqPluginConstants;
import fun.tan90.easy.binlog.plugin.rocketmq.model.RocketmqClientParams;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;

/**
 * description TODO
 * company 铁人科技
 *
 * @author 陈浩杰
 * @date 2023/11/15 14:41
 */
@Slf4j
public class RocketmqPluginService implements PluginService {

    @Override
    public boolean test(String params) {
        RocketmqClientParams rocketmqClientParams = JSONUtil.toBean(params, RocketmqClientParams.class);

        DefaultMQProducer producer = new DefaultMQProducer(rocketmqClientParams.getNamespace(), rocketmqClientParams.getProducerGroup());
        producer.setNamesrvAddr(rocketmqClientParams.getNamesrvAddr());
        producer.setSendMsgTimeout(rocketmqClientParams.getSendMsgTimeout());
        producer.setRetryAnotherBrokerWhenNotStoreOK(rocketmqClientParams.isRetryAnotherBrokerWhenNotStoreOk());
        producer.setRetryTimesWhenSendAsyncFailed(rocketmqClientParams.getRetryTimesWhenSendAsyncFailed());
        try {
            producer.start();
            Message message = new Message(RocketmqPluginConstants.TOPIC, "ping", "ping".getBytes());
            producer.sendOneway(message);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            log.error("rocketmq 连接测试失败，{}", e.getMessage());
            return false;
        } finally {
            producer.shutdown();
        }
    }
}
